package com.aotain.jupiter.domain;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.jupiter.common.URLStatTuple;

public class DomainStat implements FlatMapFunction<Tuple2<String,URLStatTuple>, Object>{

	/**
	 * 
	 */
	private static final long serialVersionUID = 720588524354211782L;

	@Override
	public void flatMap(Tuple2<String, URLStatTuple> value, Collector<Object> out)
			throws Exception {
		// TODO Auto-generated method stub
		URLStatTuple tuple = value.f1;
		String domainName = value.f0;
		
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateStr_d = sdf_d.format(date);
		
		String tableName = "METIS_DOMAIN_ALL";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		row.add(new DataColumn("DOMAIN", domainName));
		row.add(new DataColumn("REPORTTIME", dateStr_d));
		row.add(new DataColumn("HTTPNUM", tuple.getHTTPNum()));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("DOMAIN",domainName));
		filter.add(new DataColumn("REPORTTIME",dateStr_d));
		
		importtool.UpdateRowData(tableName, filter, row);
	}

}
